package com.study.kafka.custom.serialization;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * kafka生产者
 * 原理：
 * 1）生产者客户端由两个线程协调运行，主线程和Sender线程
 * 2）主线程：创建消息，通过拦截器、序列化器、分区器之后缓存到累加器(RecordAccumulator)
 * 3）Sender线程：从累加器冲获取消息发送到Kafka
 * 4)累加器作用：缓存消息（默认大小32M），以便Sender线程可以批量发送
 *
 * @author: stephen.shen
 * @create: 2019-06-12 22:08
 */
public class DemoKafkaProducer {

    private final static String brokerList = "localhost:9092";
    private final static String topic = "test-topic";

    private static Properties initConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                DemoSerializer.class.getName());
        return props;
    }

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaProducer<String, Person> producer = new KafkaProducer<>(props);
        // 构造消息
        ProducerRecord<String, Person> record = new ProducerRecord<>(topic, "kkk", new Person("zhang san", 10));
        // 异步发送
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("send success. metadata:" + metadata);
            } else {
                exception.printStackTrace();
            }
        });
        // 关闭
        producer.close();
    }

}